package net.bwie.realtime.jtp.ods.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;

/**
 * @Author: FuHe
 * @Date: 2025/6/2
 */
public class CdcDataEtlMapFunction extends RichMapFunction<String, String> {
    @Override
    public String map(String value) throws Exception {
        // 解析json字符串
        JSONObject object = JSON.parseObject(value);
        // 提取after值
        String afterValue = object.getString("after");
        // 提取before的值
        String beforeValue = object.getString("before");
        // 提取op值
        String opValue = object.getString("op");
        // 提取时间戳
        Long tsValue = object.getLong("ts_ms");
        // 提取库名
        String dbValue = object.getJSONObject("source").getString("db");
        // 提取表值
        String tableValue = object.getJSONObject("source").getString("table");
        String operatorValue;
        switch (opValue) {
            case "u":
                operatorValue = "update";
                break;
            case "r":
            case "c":
                operatorValue = "insert";
                break;
            case "d":
                operatorValue = "delete";
                break;
            default:
                operatorValue = "unknown";
        }
        // 拼接 字符串
        JSONObject newObject = new JSONObject();
        newObject.put("db_name", dbValue);
        newObject.put("table_name", tableValue);
        newObject.put("operate_type", operatorValue);
        newObject.put("ts_ms", tsValue);
        // 由于data值为json格式，需要先解析为JSONObject，在设置值
        if ("d".equals(opValue)) {
            newObject.put("operate_data", JSON.parseObject(beforeValue));
        } else {
            newObject.put("operate_data", JSON.parseObject(afterValue));
        }
        // 返回
        return newObject.toJSONString();
    }
}
